package com.codejiwei.core.graphx

import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object GraphXDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[1]").setAppName("GraphDemo")
    val sc = new SparkContext(conf)
    //创建点RDD
    val users: RDD[(VertexId, (String, String))] =
      sc.parallelize(Seq((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),
        (5L, ("franklin", "prof")), (2L, ("istoica", "prof"))))

    //创建边RDD
    val relationships: RDD[Edge[String]] =
      sc.parallelize(Seq(Edge(3L, 7L, "collab"), Edge(5L, 3L, "advisor"),
        Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi")))

    val defaultUser = ("John Doe", "Missing")

    val graph: Graph[(String, String), String] = Graph(users, relationships, defaultUser)

    val count = graph.vertices.filter {
      case (id, (name, pos)) => pos == "prof"
    }.count()

//    val count1 = graph.edges.filter(e => e.srcId > e.dstId).count()
    val count1 = graph.edges.filter {case Edge(src, dst, prop) => src > dst}.count()

    println(s"vertices count: ${count}")
    println(s"edges count: ${count1}")


    val facts = graph.triplets.map(triplet =>
      triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1
    )

    facts.collect().foreach(println)





  }
}
